package com.leyantech.ecp.agenx.session.manager;

import com.leyantech.ecp.agenx.common.ClusterVerticle;
import com.leyantech.ecp.agenx.proto.GetOnlineAssistantsByStoreRequest;
import com.leyantech.ecp.agenx.proto.GetOnlineAssistantsByStoreResponse;
import com.leyantech.ecp.agenx.proto.SessionManagerGrpc;
import com.leyantech.ecp.agenx.session.manager.di.SessionManagerModule;
import com.leyantech.ecp.agenx.session.manager.service.SessionManageService;
import com.leyantech.gaia.server.Gaia;
import com.leyantech.gaia.server.health.GrpcMethodHealthChecker;
import com.leyantech.gaia.server.health.HealthChecker;
import com.leyantech.gaia.server.health.HealthCheckers;

import com.ctrip.framework.apollo.Config;
import com.ctrip.framework.apollo.ConfigService;
import com.google.common.collect.Lists;
import com.google.inject.Inject;
import com.google.inject.Module;
import io.grpc.Status.Code;
import io.vertx.core.Promise;
import io.vertx.core.json.JsonArray;
import io.vertx.core.spi.cluster.ClusterManager;
import lombok.SneakyThrows;

import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author eahau, {@literal <eahau@foxmail.com>}
 * @date 2022-11-07.
 */
public class SessionManagerVerticle extends ClusterVerticle {

  @Inject
  private SessionManageService sessionManageService;

  @Override
  protected ClusterManager configClusterManager() {
    return null;
  }

  @Override
  protected List<Module> dependentModules() {
    return Lists.newArrayList(new SessionManagerModule());
  }

  @SneakyThrows
  private void startGaia(Promise<Void> promise) {
    log.info("SessionManageService start.");

    final Gaia gaia = Gaia.getInstance();
    gaia.addService(sessionManageService);

    final Config config = ConfigService.getConfig("warmup");
    final String storeArrayJson = config.getProperty("store", "[]");

    final JsonArray storeArray = new JsonArray(storeArrayJson);

    final AtomicInteger index = new AtomicInteger();

    final GrpcMethodHealthChecker<GetOnlineAssistantsByStoreRequest, GetOnlineAssistantsByStoreResponse> warmuper = GrpcMethodHealthChecker
        .builder(SessionManagerGrpc.METHOD_GET_ONLINE_ASSISTANTS_BY_STORE)
        .acceptCode(Code.OK)
        .setTimeoutMs(config.getLongProperty("store.timeout.ms", 50L))
        .setRequestSupplier(() -> {
          int pos = index.getAndIncrement();
          pos = pos < storeArray.size() ? pos : (pos % storeArray.size());
          return GetOnlineAssistantsByStoreRequest.newBuilder()
              .setStoreId(storeArray.getString(pos))
              .build();
        })
        .build();

    final int times = config.getIntProperty("store.times", storeArray.size());
    final HealthChecker healthChecker = HealthCheckers.repeat(warmuper, times);

    gaia.setHealthChecker(HealthCheckers.checkOnStartOnly(healthChecker));

    gaia.start();

    while (index.get() < times) {
      TimeUnit.MILLISECONDS.sleep(100L);
    }

    promise.complete();
  }

  @Override
  public void start() {
    // 启动 gaia 和预热会阻塞线程
    vertx.executeBlocking(this::startGaia)
        .onComplete(res -> {
          if (res.succeeded()) {
            log.info("SessionManageService started.");
          } else {
            log.error("SessionManageService start failed.", res.cause());
          }
        });
  }

  public static void main(String[] args) {
    new SessionManagerVerticle().startVerticle();
  }

}
